package com.fuhao.pay.mq.producer;

import com.alibaba.fastjson.JSON;
import com.fuhao.pay.constant.RabbitConstants;
import com.fuhao.pay.mq.rabbitconfig.CorrelationDataExtends;
import com.fuhao.pay.mq.rabbitconfig.SystemConfig;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import com.alibaba.fastjson.JSONObject;

import lombok.extern.slf4j.Slf4j;

import java.util.UUID;

/**
 * 生产者投递积分消息
 */
@Component
@Slf4j
public class IntegralProducer implements RabbitTemplate.ConfirmCallback ,RabbitTemplate.ReturnCallback, InitializingBean {
	@Autowired
	private RabbitTemplate rabbitTemplate;
	@Autowired
	private SystemConfig systemConfig;

	@Transactional
	public void send(String exchangeName, String routingKey, JSONObject msg) {

		String jsonString = msg.toJSONString();
		System.out.println("jsonString:" + jsonString);

		//自定义消息ID,最好使用雪花算法生产
        String msgId = UUID.randomUUID().toString();
        // 封装消息格式即消息id，给消费者
		Message message = MessageBuilder.withBody(jsonString.getBytes())
				.setContentType(MessageProperties.CONTENT_TYPE_JSON).setContentEncoding("utf-8").setMessageId(msgId)
				.build();

		// 封装CorrelationData对象，给生产者回调
		CorrelationDataExtends correlationData = new CorrelationDataExtends(msgId,message);
		correlationData.setExchange(exchangeName);
		correlationData.setRoutingKey(routingKey);
		correlationData.setMessage(message);

		//投递消息
		this.convertAndSend(exchangeName, routingKey, message,correlationData);
	}

	/**
	 * 用于实现消息发送到RabbitMQ交换器后接收ack回调。
	 * 如果消息发送确认失败就进行重试。
	 *
	 * @param correlationData
	 * @param ack
	 * @param cause
	 */
	@Override
	public void confirm(CorrelationData correlationData, boolean ack, String cause) {
		// 消息回调确认失败处理
		if (!ack && correlationData instanceof CorrelationDataExtends) {
            log.warn(">>>>>>>>>>消息投递未成功，开启重试<<<<<<<<<<<<");
			CorrelationDataExtends correlationDataExtends = (CorrelationDataExtends) correlationData;

			//消息发送失败,就进行重试，重试过后还不能成功就记录到数据库
			if (correlationDataExtends.getRetryCount() < systemConfig.getMqRetryCount()) {
				log.info("MQ消息发送失败，消息重发，消息ID：{}，重发次数：{}，消息体:{}", correlationDataExtends.getId(),
						correlationDataExtends.getRetryCount(), JSON.toJSONString(correlationDataExtends.getMessage()));

				// 将重试次数加一
				correlationDataExtends.setRetryCount(correlationDataExtends.getRetryCount() + 1);

				// 重发发消息
				this.convertAndSend(correlationDataExtends.getExchange(), correlationDataExtends.getRoutingKey(),
						correlationDataExtends.getMessage(), correlationDataExtends);
			} else {
				//消息重试发送失败,将消息放到数据库等待补发
				log.warn("MQ消息重发失败，消息入库，消息ID：{}，消息体:{}", correlationData.getId(),
						JSON.toJSONString(correlationDataExtends.getMessage()));

				// TODO 保存消息到数据库
			}
		} else {
			log.info("消息发送成功,消息ID:{}", correlationData.getId());
		}
	}

	/**
	 * 用于实现消息发送到RabbitMQ交换器，但无相应队列与交换器绑定时的回调。
	 * 在脑裂的情况下会出现这种情况
	 */
	@Override
	public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
		log.error("MQ消息发送失败，replyCode:{}, replyText:{}，exchange:{}，routingKey:{}，消息体:{}",
				replyCode, replyText, exchange, routingKey, JSON.toJSONString(message.getBody()));

		// TODO 保存消息到数据库
	}

	@Override
	public void afterPropertiesSet() throws Exception {
		rabbitTemplate.setConfirmCallback(this);
		rabbitTemplate.setReturnCallback(this);
	}

	/**
	 * 消息相关数据（消息ID）
	 *
	 * @param message
	 * @return
	 */
	private CorrelationDataExtends correlationData(Object message) {

		return new CorrelationDataExtends(UUID.randomUUID().toString(), message);
	}


	/**
	 * 发送消息
	 *
	 * @param exchange        交换机名称
	 * @param routingKey      路由key
	 * @param message         消息内容
	 * @param correlationData 消息相关数据（消息ID）
	 * @throws AmqpException
	 */
	private void convertAndSend(String exchange, String routingKey, final Object message, CorrelationData correlationData) throws AmqpException {
		try {
			rabbitTemplate.convertAndSend(exchange, routingKey, message, correlationData);
		} catch (Exception e) {
			log.error("MQ消息发送异常，消息ID：{}，消息体:{}, exchangeName:{}, routingKey:{}",
					correlationData.getId(), JSON.toJSONString(message), exchange, routingKey, e);

			// TODO 保存消息到数据库
		}
	}
}